package com.piece.core.kafka.consumer;

import com.piece.core.framework.constant.RegisterConstants;
import com.piece.core.framework.support.dto.PayloadDTO;
import com.piece.core.framework.util.message.MessageFactory;
import com.piece.core.framework.util.string.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import java.util.List;

@Slf4j
public class KafkaMessageConsumer {

    private MessageFactory messageFactory;

    public KafkaMessageConsumer(MessageFactory messageFactory) {
        this.messageFactory = messageFactory;
    }

    @KafkaListener(topics = { RegisterConstants.TOPIC }, groupId = "${spring.application.name}")
    public void handleMessageReceive(List<ConsumerRecord<String, String>> records, Acknowledgment ack) throws Exception {
        for (ConsumerRecord<String, String> record : records) {
            String payloadJson = record.value();
            PayloadDTO payload = JsonUtil.toBean(payloadJson, PayloadDTO.class);
            try {
                messageFactory.getOperator(payload.getOperator()).doReceive(payload);
                // 手动删除消息
                ack.acknowledge();
            } catch (Exception e) {
                // 手动删除消息
                ack.acknowledge();
            }
        }
    }

    @KafkaListener(topics = { RegisterConstants.DEAD_LETTER_TOPIC })
    public void handleDLTMessageReceive(ConsumerRecord<String, String> record) {
        log.debug("收到死信消息:topic({}),record({})", RegisterConstants.DEAD_LETTER_TOPIC, record);
    }
}
